package pl.cezary.butler.walili.client;

import pl.cezary.butler.walili.common.IServerFacadeRemote;

import pl.cezary.butler.walili.common.Task;
import java.rmi.RemoteException;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import pl.cezary.butler.walili.client.config.ClientConfiguration;

/**
 * Klasa odpowiedzialna za obsługę zadań znajdujących się w kolejce. Klasa nie
 * komunikuje się bezpośrednio z bazą danych
 *
 * @author Cezary Butler
 */
public class TaskFetchingThread extends Thread {

    private static final Logger log = Logger.getLogger(TaskFetchingThread.class.getName());
    private static final ClientConfiguration CONFIG = ClientConfiguration.getInstance();
    private static final int WORKING_THREADS = CONFIG.getClientWorkingThreads();
    private static final int TASKS_TO_FETCH = CONFIG.getClientTasksQueueLength();
    private static final int MILIS_TO_WAIT = CONFIG.getClientTaskFetchDelay();
    private final ThreadGroup excecutedTasks = new ThreadGroup("MainTaskExecutorGroup");
    private final IServerFacadeRemote fasada;

    public TaskFetchingThread(IServerFacadeRemote fasada) {
        super(TaskFetchingThread.class.getSimpleName());
        assert (fasada != null);
        this.fasada = fasada;
    }
    private int threadNo = 0;

    @Override
    public void run() {
        final ExecutorService srv = Executors.newFixedThreadPool(WORKING_THREADS, new ThreadFactory() {

            @Override
            public Thread newThread(Runnable r) {
                final Thread th = new Thread(excecutedTasks, r, "TaskExecutingThread" + ++threadNo);
                th.setPriority(MIN_PRIORITY);
                return th;
            }
        });

        try {
            log.info("Task fetching was initiated");

            while (!isInterrupted()) {
                try {
                    int currentTasks;
                    if ((currentTasks = TaskExecutionUnit.running()) < TASKS_TO_FETCH) {
                        final int tryingToFetch = TASKS_TO_FETCH - currentTasks;
                        if(log.isLoggable(Level.FINE))
                            log.log(Level.FINE, "Fetching new {0} tasks", (tryingToFetch));
                        int cntr = 0;
                        for (Task zad : fasada.getTasks(tryingToFetch)) {
                            if (zad == null) {
                                break;
                            }
                            ++cntr;
                            srv.execute(new TaskExecutionUnit(zad, fasada));
                        }
                        if (cntr < tryingToFetch) {
                            if(log.isLoggable(Level.INFO))
                                log.log(Level.INFO, "Currently {0} tasks are working tried to fetch {1}, received {2}",
                                        new Object[]{currentTasks, tryingToFetch, cntr});
                        }
                    }

                    TimeUnit.MILLISECONDS.sleep(MILIS_TO_WAIT);

                } catch (InterruptedException ex) {
                    log.log(Level.INFO, "Task fetching thread has been interrupted.");
                    break;
                }
            }
        } catch (RemoteException ex) {
            log.log(Level.SEVERE, "RMI communication error - look into server log.", ex);
        } finally {
            log.info("Task executor service is being shutdown");
            try {
                srv.shutdown();
                if (!srv.awaitTermination(500, TimeUnit.MILLISECONDS)) {
                    final List<Runnable> leftBehind = srv.shutdownNow();
                    log.log(Level.INFO, "{0} tasks have been left behind", leftBehind.size());
                }
                log.info("Task executor service has been shutdown");
            } catch (InterruptedException ex) {
                log.log(Level.SEVERE, "Waiting for executor service termination was interrupted", ex);
            }
        }
    }
}
